package platform

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"gitee.com/Frank098/devops-app-mgt/global"
	"github.com/gorilla/websocket"
	"io"
	v1 "k8s.io/api/core/v1"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/tools/remotecommand"
	"net/http"
	"sync"
	"time"
)

type terminal struct{}

// http升级websocket协议的配置
var wsUpgrader = websocket.Upgrader{
	// 允许所有CORS跨域请求
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

// WsMessage websocket消息
type WsMessage struct {
	MessageType int
	Data        []byte
}

// WsConnection 封装websocket连接
type WsConnection struct {
	wsSocket *websocket.Conn // 底层websocket
	inChan   chan *WsMessage // 读取队列
	outChan  chan *WsMessage // 发送队列

	mutex     sync.Mutex // 避免重复关闭管道
	isClosed  bool
	closeChan chan byte // 关闭通知
}

// 流式处理器
type streamHandler struct {
	wsConn      *WsConnection
	resizeEvent chan remotecommand.TerminalSize
}

// web终端发来的包
// ws://127.0.0.1:8090/apis/apps/v1/pods/ssh?pod_name=web-01-75dc579c5b-k9qdw
// {"type":"input","input":"ls -l","rows":0,"cols":0}
// {"type":"input","input":"\r","rows":0,"cols":0}
type xtermMessage struct {
	MsgType string `json:"type"`  // 类型:resize客户端调整终端, input客户端输入
	Input   string `json:"input"` // msgtype=input情况下使用
	Rows    uint16 `json:"rows"`  // msgtype=resize情况下使用
	Cols    uint16 `json:"cols"`  // msgtype=resize情况下使用
}

// 读取协程
func (wsConn *WsConnection) wsReadLoop() {
	var (
		msgType int
		data    []byte
		msg     *WsMessage
		err     error
	)
	for {
		// 读一个message
		if msgType, data, err = wsConn.wsSocket.ReadMessage(); err != nil {
			goto ERROR
		}
		msg = &WsMessage{
			msgType,
			data,
		}
		// 放入请求队列
		select {
		case wsConn.inChan <- msg:
		case <-wsConn.closeChan:
			goto CLOSED
		}
	}
ERROR:
	wsConn.WsClose()
CLOSED:
}

// 发送协程
func (wsConn *WsConnection) wsWriteLoop() {
	var (
		msg *WsMessage
		err error
	)
	for {
		select {
		// 取一个应答
		case msg = <-wsConn.outChan:
			// 写给websocket
			if err = wsConn.wsSocket.WriteMessage(msg.MessageType, msg.Data); err != nil {
				goto ERROR
			}
		case <-wsConn.closeChan:
			goto CLOSED
		}
	}
ERROR:
	wsConn.WsClose()
CLOSED:
}

func (t *terminal) InitWebsocket(resp http.ResponseWriter, req *http.Request) (wsConn *WsConnection, err error) {
	var (
		wsSocket *websocket.Conn
	)
	// 应答客户端告知升级连接为websocket
	if wsSocket, err = wsUpgrader.Upgrade(resp, req, nil); err != nil {
		return
	}
	wsConn = &WsConnection{
		wsSocket:  wsSocket,
		inChan:    make(chan *WsMessage, 1000),
		outChan:   make(chan *WsMessage, 1000),
		closeChan: make(chan byte),
		isClosed:  false,
	}

	// 读协程
	go wsConn.wsReadLoop()
	// 写协程
	go wsConn.wsWriteLoop()

	return
}

// 发送消息
func (wsConn *WsConnection) WsWrite(messageType int, data []byte) (err error) {
	select {
	case wsConn.outChan <- &WsMessage{messageType, data}:
	case <-wsConn.closeChan:
		err = errors.New("websocket closed")
	}
	return
}

// 读取消息
func (wsConn *WsConnection) WsRead() (msg *WsMessage, err error) {
	select {
	case msg = <-wsConn.inChan:
		return
	case <-wsConn.closeChan:
		err = errors.New("websocket closed")
	}
	return
}

// 关闭连接
func (wsConn *WsConnection) WsClose() {
	wsConn.wsSocket.Close()

	wsConn.mutex.Lock()
	defer wsConn.mutex.Unlock()
	if !wsConn.isClosed {
		wsConn.isClosed = true
		close(wsConn.closeChan)
	}
}

// Next executor回调获取web是否resize
func (handler *streamHandler) Next() (size *remotecommand.TerminalSize) {
	ret := <-handler.resizeEvent
	size = &ret
	return
}

// executor回调读取web端的输入
func (handler *streamHandler) Read(p []byte) (size int, err error) {
	var (
		msg      *WsMessage
		xtermMsg xtermMessage
	)

	// 读web发来的输入
	if msg, err = handler.wsConn.WsRead(); err != nil {
		return
	}

	// 解析客户端请求
	if err = json.Unmarshal(msg.Data, &xtermMsg); err != nil {
		return
	}

	//web ssh调整了终端大小
	if xtermMsg.MsgType == "resize" {
		// 放到channel里，等remotecommand executor调用我们的Next取走
		handler.resizeEvent <- remotecommand.TerminalSize{Width: xtermMsg.Cols, Height: xtermMsg.Rows}
	} else if xtermMsg.MsgType == "input" { // web ssh终端输入了字符
		// copy到p数组中
		size = len(xtermMsg.Input)
		copy(p, xtermMsg.Input)
	}
	return
}

// executor回调向web端输出
func (handler *streamHandler) Write(p []byte) (size int, err error) {
	var (
		copyData []byte
	)

	// 产生副本
	copyData = make([]byte, len(p))
	copy(copyData, p)
	size = len(p)
	err = handler.wsConn.WsWrite(websocket.TextMessage, copyData)
	return
}

func (t *terminal) PodWsSsh(PodName string, resp http.ResponseWriter, req *http.Request) {
	var (
		wsConn   *WsConnection
		restConf *rest.Config
		Req      *rest.Request
		executor remotecommand.Executor
		handler  *streamHandler
		err      error
	)

	podsApi, _, err := K8s.Plat.Pods().GetPod(PodName)
	if err != nil {
		global.Logger.Errorf("获取Pods的API接口信息失败：" + err.Error())
	}

	// 得到websocket长连接
	if wsConn, err = t.InitWebsocket(resp, req); err != nil {
		return
	}

	// 获取k8s rest client配置
	if restConf, err = clientcmd.BuildConfigFromFlags("", global.KubeConfigSetting.Path+"\\.kube\\config"); err != nil {
		goto END
	}

	Req = K8s.ClientSet.CoreV1().RESTClient().Post().
		Resource("pods").
		Name(podsApi.Name).
		Namespace(podsApi.Namespace).
		SubResource("exec").
		VersionedParams(&v1.PodExecOptions{
			Container: podsApi.Spec.Containers[0].Name,
			Command:   []string{"bash"},
			Stdin:     true,
			Stdout:    true,
			Stderr:    true,
			TTY:       true,
		}, scheme.ParameterCodec)

	// 创建到容器的连接
	if executor, err = remotecommand.NewSPDYExecutor(restConf, "POST", Req.URL()); err != nil {
		goto END
	}

	// 配置与容器之间的数据流处理回调
	handler = &streamHandler{wsConn: wsConn, resizeEvent: make(chan remotecommand.TerminalSize)}

	if err = executor.StreamWithContext(context.TODO(), remotecommand.StreamOptions{
		Stdin:             handler,
		Stdout:            handler,
		Stderr:            handler,
		TerminalSizeQueue: handler,
		Tty:               true,
	}); err != nil {
		goto END
	}
	return

END:
	wsConn.WsClose()
}

type ClientMsg struct {
	Type string
	Data *PodsLogsParams
}

type PodsLogsParams struct {
	ContainerName string
	Timestamps    bool
	SinceSeconds  int
	TailLines     int
}

func (t *terminal) PodWsLogs(podName, namespace, containerName string, timestamps bool, sinceSeconds, tailLines int, resp http.ResponseWriter, req *http.Request) {
	var (
		wsConn  *WsConnection
		Req     *rest.Request
		handler *streamHandler
		err     error
	)

	// 得到websocket长连接
	if wsConn, err = t.InitWebsocket(resp, req); err != nil {
		return
	}

	handler = &streamHandler{wsConn: wsConn, resizeEvent: make(chan remotecommand.TerminalSize)}

	if sinceSeconds == 0 && tailLines == 0 {
		firstSendReq := K8s.ClientSet.CoreV1().Pods(namespace).GetLogs(podName,
			&v1.PodLogOptions{})
		firstSendLogs, err := firstSendReq.Stream(context.TODO())
		if err != nil {
			handler.Write([]byte(err.Error()))
			time.Sleep(1 * time.Second)
			goto END
		}

		defer firstSendLogs.Close()

		// 将Response.body 写入到缓存区,目的为了转换成string类型
		buf := new(bytes.Buffer)
		_, err = io.Copy(buf, firstSendLogs)
		if err != nil {
			handler.Write([]byte(err.Error()))
			time.Sleep(1 * time.Second)
			goto END
		}

		handler.Write([]byte(buf.String()))
		time.Sleep(2 * time.Second)

		continueSendReq := K8s.ClientSet.CoreV1().Pods(namespace).GetLogs(podName,
			&v1.PodLogOptions{Follow: true, TailLines: func(i int64) *int64 { return &i }(1)})
		continueSendLogs, err := continueSendReq.Stream(context.TODO())
		if err != nil {
			handler.Write([]byte(err.Error()))
			time.Sleep(1 * time.Second)
			goto END
		}
		t.MessageProcess(handler, continueSendLogs)
	} else {
		Req = t.PodsLogsRead(podName, namespace, containerName, timestamps, sinceSeconds, tailLines)

		reader, err := Req.Stream(context.TODO())
		if err != nil {
			handler.Write([]byte(err.Error()))
			time.Sleep(1 * time.Second)
			goto END
		}
		t.MessageProcess(handler, reader)
	}

END:
	wsConn.WsClose()
}

func (t *terminal) MessageProcess(handler *streamHandler, reader io.ReadCloser) {
	go io.Copy(handler, reader)
	buf := make([]byte, 256)
	for {
		_, err := handler.Read(buf)
		if err != nil {
			_, _ = handler.Write([]byte(err.Error()))
			break
		}
		_, _ = handler.Write(buf)
	}
	return
}

func (t *terminal) PodsLogsRead(podName, namespace, containerName string, timestamps bool, sinceSeconds, tailLines int) (req *rest.Request) {
	if tailLines != 0 {
		req = K8s.ClientSet.CoreV1().Pods(namespace).GetLogs(podName, &v1.PodLogOptions{
			Container:  containerName,
			Follow:     true,
			Timestamps: timestamps,
			TailLines:  func(i int64) *int64 { return &i }(int64(tailLines)),
		})
		return req
	} else if sinceSeconds != 0 {
		req = K8s.ClientSet.CoreV1().Pods(namespace).GetLogs(podName, &v1.PodLogOptions{
			Container:    containerName,
			Follow:       true,
			SinceSeconds: func(i int64) *int64 { return &i }(int64(sinceSeconds)),
			Timestamps:   timestamps,
		})
		return req
	} else {
		req = K8s.ClientSet.CoreV1().Pods(namespace).GetLogs(podName, &v1.PodLogOptions{
			Container:  containerName,
			Follow:     true,
			Timestamps: timestamps,
		})
		return req
	}
}
